package cn.basttg.demo.zookeeper;

import java.io.UnsupportedEncodingException;
import java.util.List;

import net.sf.json.JSONObject;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZkMcu extends Thread implements Watcher {
	private static final String DATA_NODE = "operation";
	private ZooKeeper zk = null;
	private Integer mutex;
	private String root;
	private boolean running = true; 

	synchronized public void process(WatchedEvent event) {
		synchronized (mutex) {
			// System.out.println("Process: " + event.getType());
			mutex.notify();
		}
	}

	@Override
	public void run() {
		while(running) {
			try {
				String data = consume();
				System.out.println("Consume Data: " + data);
            } catch (KeeperException e){
            	System.out.println("Consume error: ");
            	e.printStackTrace();
            } catch (InterruptedException e){

            }
		}
	}
	
	public void stopRunning() {
		running = false;
	}

	/**
	 * Constructor of producer-consumer queue
	 * @param connectString
	 * @param rootPath
	 */
	ZkMcu(String connectString, String rootPath) {
		if (zk == null) {
			try {
				System.out.println("Starting ZK:");
				zk = new ZooKeeper(connectString, 3000, this);
				mutex = new Integer(-1);
				System.out.println("Finished starting ZK: " + zk);
			} catch (Exception e) {
				System.out.println(e.toString());
				zk = null;
			}
		}
		// else mutex = new Integer(-1);

		this.root = rootPath;
		// Create ZK node name
		if (zk != null) {
			try {
				Stat s = zk.exists(root, false);
				if (s == null) {
					zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}
			} catch (KeeperException e) {
				System.out.println("Keeper exception when instantiating queue: " + e.toString());
			} catch (InterruptedException e) {
				System.out.println("Interrupted exception");
			}
		}
	}

	/**
	 * Remove first element from the queue.
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	String consume() throws KeeperException, InterruptedException {
		String value = "";
		Stat stat = null;

		// Get the first element available
		while (true) {
			synchronized (mutex) {
				List<String> list = zk.getChildren(root, true);
				if (list.size() == 0) {
					System.out.println("Going to wait");
					mutex.wait();
				} else {
					int len = DATA_NODE.length();
					String minPath = list.get(0);
					Integer min = new Integer(list.get(0).substring(len));
					for (String s : list) {
						Integer tempValue = new Integer(s.substring(len));
						if (tempValue < min) {
							min = tempValue;
							minPath = s;
						}
					}
					System.out.println("Temporary value: " + root + "/" + minPath);
					byte[] b = zk.getData(root + "/" + minPath, false, stat);
					zk.delete(root + "/" + minPath, 0);
					try {
						value = new String(b, "UTF-8");
						
						JSONObject operation = JSONObject.fromObject(value);
						String enterprise = operation.getString("enterprise");
						String meetingId = operation.getString("meetingId");
						String data = operation.getString("data");
						String type = operation.getString("type");
						if("CREATE_MEETING".equals(type)) {
							zk.create("/webmcc/" + enterprise + "/" + meetingId, data.getBytes("UTF-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
						}
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}

					return value;
				}
			}
		}
	}

	public static void main(String args[]) {
		String connectionString = "172.16.160.196:2181";
		String rootPath = "/app1";
		int max = 100;

		ZkMcu mcu = new ZkMcu(connectionString, rootPath);
		for (int i = 0; i < max; i++) {
			try {
				String data = mcu.consume();
				System.out.println("Consume Item" + i + ": " + data);
            } catch (KeeperException e){
                i--;
            } catch (InterruptedException e){

            } catch (Exception e) {
				System.out.println("Consume error: " + i);
			}
		}
	}
}

